package com.huan.hotitems;


import com.huan.bean.ItemViewCount;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

//自定义全窗口函数
public class WindowItemCountResult implements WindowFunction<Long, ItemViewCount, Tuple, TimeWindow> {

    @Override
    public void apply(Tuple tuple, TimeWindow window, Iterable<Long> input, Collector<ItemViewCount> out) throws Exception {
        /**
         * 因为输出类型是 ItemViewCount 类型，是根据商品 id 累计点击 到时间后触发窗口操作，
         * 则 列出  ItemViewCount 类的字段即可
         */
        Long ItemId = tuple.getField( 0 ); // 获取id
        Long windowEnd = window.getEnd(); //获取窗口结束时间
        Long count = input.iterator().next(); //获取点击
        out.collect( new ItemViewCount(ItemId,windowEnd,count) );

    }
}
